package com.wenge.datagroup.storage.future

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.wenge.datagroup.storage.bean.Field
import com.wenge.datagroup.storage.dataprocess.{MultilingualLanguage, MultilingualTranslation}
import com.wenge.datagroup.storage.util.{DruidDataSource, MethodMatch}
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.util.Time.now
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast

import java.lang.{Float => JavaFloat, Integer => JavaInt, Long => JavaLong}
import java.util.Date
import java.text.SimpleDateFormat
import java.time.format.DateTimeFormatter
import java.util
import java.util.{LinkedList => JavaLinkedList}
import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters.asScalaBufferConverter
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.runtime.BoxedUnit

/**
 * @author
 * Depp
 */
class AsyncProcess(
                    druidDataSourceBro: Broadcast[DruidDataSource],
                    multilingualLanguageBro: Broadcast[MultilingualLanguage],
                    multilingualTranslationBro: Broadcast[MultilingualTranslation]) {
  private val logger: Logger = Logger.getLogger(classOf[AsyncProcess])
  private val map: ConcurrentHashMap[String, Any] = new ConcurrentHashMap[String, Any]()
  private val EMPTY_LIST = new util.ArrayList[String]
  private val EMPTY_ARRAY = new Array[String](0)
  private val EMPTY_JSONOBJECT = new JSONObject
  private val EMPTY_JSONARRAY = new JSONArray
  private val methodMatch = new MethodMatch
  private val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

  def runAsync(
                fields: JavaLinkedList[Field],
                orgRecord: JSONObject): ConcurrentHashMap[String, Any] = {
    val scalaFields: List[Field] = fields.asScala.toList

    val setDefaultValueTask: Future[Unit] = Future {
      setDefaultValueFuture(map, scalaFields)
    }

    val setTimeFormatAndOrgValueTasks = setTimeFormatAndOrgValueFuture(map, orgRecord, scalaFields)

    val specialTransformTasks = specialTransformFuture(map, orgRecord, scalaFields)

    val combined: Future[Unit] =
      for (n1 <- setTimeFormatAndOrgValueTasks; n2 <- specialTransformTasks) yield null

    while (!setDefaultValueTask.isCompleted || !combined.isCompleted) {
      Thread.sleep(5)
      logger.info("waiting all futures completed...")
    }

    map
  }

  private[this] def setTimeFormatAndOrgValueFuture(
                                                    map: ConcurrentHashMap[String, Any],
                                                    orgRecord: JSONObject,
                                                    scalaFields: List[Field]): Future[List[Unit]] = {
    val setTimeFormatAndOrgValueFutures: List[Future[Unit]] = scalaFields
      .filter {
        field => {
          StringUtils.isNotBlank(field.getFinalField) && (StringUtils.isBlank(
            field.getClazz) && StringUtils.isBlank(field.getMethodName) && (StringUtils
            .isNotBlank(field.getOrgField) || StringUtils.isNotBlank(field.getFormat)))
        }
      }
      .map(
        timeFormatAndOrgValueField =>
          Future {

            logger.info("执行此任务的线程是：" +Thread.currentThread().getName+",time="+sdf.format(new Date()))
            val orgFieldType = timeFormatAndOrgValueField.getOrgFieldType
            val orgField = timeFormatAndOrgValueField.getOrgField
            val finalField = timeFormatAndOrgValueField.getFinalField
            val format = timeFormatAndOrgValueField.getFormat

            var orgValue: Any = null
            orgFieldType match {
              case "int" =>
                orgValue = orgRecord.getInteger(orgField)
              case "long" =>
                orgValue = orgRecord.getLong(orgField)
              case "arraylist" | "jsonarray" =>
                orgValue = orgRecord.getJSONArray(orgField)
              case "jsonobject" =>
                orgValue = orgRecord.getJSONObject(orgField)
              case "float" =>
                orgValue = orgRecord.getFloatValue(orgField)
              case _ =>
                orgValue = orgRecord.getString(orgField)
            }
            if (StringUtils.isNotBlank(format)) { // 判断对应原字段是否有值
              if (StringUtils.isBlank(orgValue.asInstanceOf[String])) { // 取当前时间，秒
                if ("llt".equals(orgField) || "lrt".equals(orgField)) {
                  map.put(finalField, "")
                } else {
                  map.put(finalField, new Date(now).getTime / 1000)
                }
              } else
                try {
                  val sdf = new SimpleDateFormat(format)
                  val time = sdf.parse(orgValue.asInstanceOf[String]).getTime
                  //val sdf = DateTimeFormatter.ofPattern(format)
                  //logger.info("转换前:"+orgValue+"——转换后:"+time)
                  map.put(finalField, time / 1000)
                } catch {
                  case e: Exception =>
                    e.printStackTrace()
                    logger.error(e)
                }
            } else { // 无，旧字段有值，取旧字段值 无值，走默认值逻辑
              if (orgValue == null || orgValue.equals("")) {
                setDefaultValue(map, timeFormatAndOrgValueField, finalField)
              }
              else map.put(finalField, orgValue)
            }
            logger.info(
              s"setTimeFormatAndOrgValueFuture ${timeFormatAndOrgValueField.getFinalField} ")
          })
    Future.sequence(setTimeFormatAndOrgValueFutures)
  }

  private[this] def setDefaultValueFuture(
                                           map: ConcurrentHashMap[String, Any],
                                           scalaFields: List[Field]): Unit = {
    for (
      field <- scalaFields if StringUtils.isNotBlank(field.getFinalField) && StringUtils.isBlank(
        field.getOrgField) && StringUtils.isBlank(field.getFormat) && StringUtils.isBlank(
        field.getClazz) && !field.getMethodName.equals("continue")
    ) {
      logger.info("执行此任务的线程是：" +Thread.currentThread().getName+",time="+sdf.format(new Date()))

      setDefaultValue(map, field, field.getFinalField)
      logger.info(s"setDefaultValueFuture ${field.getFinalField} ")
    }
  }

  private[this] def specialTransformFuture(
                                            map: ConcurrentHashMap[String, Any],
                                            orgRecord: JSONObject,
                                            scalaFields: List[Field]): Future[List[Unit]] = {
    val specialTransformFutures: List[Future[Unit]] = scalaFields
      .filter {
        field =>
          StringUtils.isNotBlank(field.getFinalField) && StringUtils
            .isNotBlank(field.getClazz) && StringUtils.isNotBlank(field.getMethodName)
      }
      .map(
        specialTransformField =>
          Future {
            logger.info(specialTransformField.getMethodName+"执行此任务的线程是：" +Thread.currentThread().getName+",time="+sdf.format(new Date()))

            /* println(
               s"${specialTransformField.getMethodName} 执行此任务的线程是： ${Thread.currentThread().getName}")*/
            specialTransform(
              orgRecord,
              map,
              specialTransformField,
              specialTransformField.getOrgField,
              specialTransformField.getFinalField,
              druidDataSourceBro.value,
              multilingualLanguageBro.value,
              multilingualTranslationBro.value,
              specialTransformField.getMethodName,
              specialTransformField.getPreValue
            )
            logger.info(
              s"${specialTransformField.getFinalField} ${specialTransformField.getMethodName} ")
          })
    Future.sequence(specialTransformFutures)
  }

  private[this] def setDefaultValue(
                                     map: ConcurrentHashMap[String, Any],
                                     field: Field,
                                     finalField: String): Unit = {
    val finalFieldType = field.getFinalFieldType
    if (StringUtils.isBlank(finalFieldType))
      map.put(finalField, field.getDefultValue)
    else
      finalFieldType match {
        case "int" =>
          map.put(
            finalField,
            if (StringUtils.isBlank(field.getDefultValue)) 0
            else JavaInt.valueOf(field.getDefultValue)
          )
        case "long" =>
          map.put(
            finalField,
            if (StringUtils.isBlank(field.getDefultValue)) 0L
            else JavaLong.valueOf(field.getDefultValue)
          )
        case "float" =>
          map.put(
            finalField,
            if (StringUtils.isBlank(field.getDefultValue)) 0.0f
            else JavaFloat.valueOf(field.getDefultValue))
        case "arraylist" =>
          map.put(finalField, EMPTY_LIST)
        case "array" =>
          map.put(finalField, EMPTY_ARRAY)
        case "jsonobject" =>
          map.put(finalField, EMPTY_JSONOBJECT)
        case "jsonarray" =>
          map.put(finalField, EMPTY_JSONARRAY)
        case _ =>
      }
  }

  private[this] def specialTransform(
                                      orgRecord: JSONObject,
                                      map: ConcurrentHashMap[String, Any],
                                      field: Field,
                                      orgField: String,
                                      finalField: String,
                                      druidDataSource: DruidDataSource,
                                      multilingualLanguage: MultilingualLanguage,
                                      multilingualTranslation: MultilingualTranslation,
                                      methodName: String,
                                      preValue: String): ConcurrentHashMap[String, Any] = {
    val invoke =
      methodMatch.transformMethodMatch(
        methodName,
        druidDataSource,
        multilingualLanguage,
        multilingualTranslation,
        orgRecord,
        map,
        orgField,
        finalField,
        preValue)
    if (invoke != null && (!invoke.isInstanceOf[BoxedUnit]))
      invoke match {
        case map: ConcurrentHashMap[String, Any] =>
          return map
        case _ => map.put(finalField, invoke)
      }
    else setDefaultValue(map, field, finalField)
    map
  }
}
